package com.biao.thread.myrequestmerge.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * 商品信息服务类
 */
@Service
public class CommodityService {

    /**
     * 定义接收用户请
     */
    class Request{
        private String movieCode;

        /**
         * CompletableFuture里面的参数就是需要返回的返回值类型
         */
        private CompletableFuture<Map<String, Object>> future;
    }

    /**
     * 队列用来存请求
     */
    BlockingQueue<Request> queue = new LinkedBlockingDeque<>();

    /**
     * 调用远程资源
     */
    @Autowired
    private QueryServiceRemoteCall queryServiceRemoteCall;

    /**
     * 查询商品信息----原始版本，一次请求，一次查询
     * @param movieCode
     * @return
     */
    public Map<String,Object> queryCommodity(String movieCode) {

        /**
         * 一个请求，一个查询的方式，对于高并发，显然是浪费的
         */
        return queryServiceRemoteCall.queryCommodityByCode(movieCode);
    }


    /**
     * 让定时任务在类初始化的时候就进行加载
     */
    @PostConstruct
    public void init(){
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
        /**
         * scheduleAtFixedRate方法解析：
         * 执行方法体
         * 0毫秒立即执行，
         * 每10毫秒执行一次，
         * 时间是按毫秒来定
         */
        scheduledExecutorService.scheduleAtFixedRate(()->{
            int size = queue.size();
            if(size == 0){
                return;
            }

            ArrayList<Request> requests = new ArrayList<>();
            for (int i = 0; i < size; i++) {
                Request request = queue.poll();
                requests.add(request);
            }

            System.out.println("批量处理了：" + size + "条请求");

            // id = 1, id = 2, id = 3 , id = 4, .. ---> ids = {1,2,3,4}
            List<String> codes = new ArrayList<>();
            for(Request each: requests){
                codes.add(each.movieCode);
            }

            List<Map<String, Object>> responses = queryServiceRemoteCall.queryCommodityByCodeBatch(codes);

            // 结果集完成-----》把请求分发给每一个具体的Request....
            HashMap<String, Map<String, Object>> responseMap = new HashMap<>();
            for(Map<String, Object> response: responses){
                String code = response.get("code").toString();
                responseMap.put(code, response);
            }

            /**
             * 如何返回请求？
             * return ??? ----线程间能return吗？？？----》JDK---Callable---Future<T>-----T就是返回的结果
             * 使用get()来获取结果，它是阻塞直到有结果为止！！！！
             *
             */
            for(Request request: requests){
                Map<String, Object> result = responseMap.get(request.movieCode);
                // 通知到使用future的线程，有结果可以用了
                request.future.complete(result);
            }

        },0, 10, TimeUnit.MILLISECONDS);
    }



    /**
     * 查询商品信息----改进版本，合并请求
     * @param movieCode
     * @return
     * @throws ExecutionException
     */
    public Map<String,Object> queryCommodityV2(String movieCode) throws ExecutionException, InterruptedException {

        /**
         * 请求合并----》1000次并发同时执行
         * 拦截请求----》如何拦截？？？
         */
        Request request = new Request();
        request.movieCode = movieCode;


        /**
         * 这里让每一个线程都有自己的future，并等待定时任务完成后的结果才返回给用户
         */
        CompletableFuture<Map<String, Object>> future = new CompletableFuture<>();
        request.future = future;

        queue.add(request);

        // 直到有结果才返回，我们可以简单认为它是在死循环，直到有人给他放了数据才跳出死等状态，所以放数据给他的是谁都可以，这里是使用了定时任务来放数据而已
        return future.get();
    }

}
